package net.bwie.realtime.jtp.dwd.log.lx2.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String,String,String> {
        private transient ValueState<String> firstDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        firstDateState =
                getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDataState", String.class));
    }

    @Override
    public void processElement(String s, KeyedProcessFunction<String, String, String>.Context context, Collector<String> collector) throws Exception {
        // 获取状态值
        String firstDate = firstDateState.value();

        // 解析字段值
        JSONObject jsonObject = JSON.parseObject(s);

        // 获取 isnew 值
        String isNewValue = jsonObject.getJSONObject("common").getString("isNew");
        // 获取 ts 时间值
        Long tsValue = jsonObject.getLong("ts");

        // isNew = 1  -->   firstDate = null，  firstDate = 有值
        if("1".equals(isNewValue)){
            if (null == firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }else{
                //状态值 （首次询问日期） 等于数据日期 不做处理
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    jsonObject.getJSONObject("common").put("isNew", "0");
                }
            }
        }
        // isNew = 0  -->   firstDate = null，  firstDate = 有值
        else{
            if (null != firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 86400000, DateTimeUtil.DATE_FORMAT));
            }
        }

        collector.collect(jsonObject.toJSONString());

    }
}
